package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.util.MySinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/2/8 18:34
 */
public class DWSVisitorStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DWSVisitorStatsApp().init(2,
                                      "DWSVisitorStatsApp",
                                      "dwd_page_log", "dwm_uv", "dwm_user_jump_detail");
        
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> sourceStreams) {
        // 1. 解析3个流,并合并成一个流
        final DataStream<VisitorStats> visitorStatsDataStream = parseStreamsAndUnionOne(sourceStreams);
        // 2. 按照维度进行聚合
        final SingleOutputStreamOperator<VisitorStats> statsStream = aggregateByDims(visitorStatsDataStream);
        // 3. 数据sink掉 clickhouse
        sink2Clickhouse(statsStream);
    }
    
    private void sink2Clickhouse(SingleOutputStreamOperator<VisitorStats> statsStream) {
        statsStream.print();
        statsStream.addSink(MySinkUtil.getClickHouseSink("gmall2021", "visitor_stats_2021", VisitorStats.class));
    }
    
    /**
     * 进行聚合
     *
     * @param visitorStatsDataStream
     * @return
     */
    private SingleOutputStreamOperator<VisitorStats> aggregateByDims(DataStream<VisitorStats> visitorStatsDataStream) {
        final SingleOutputStreamOperator<VisitorStats> result = visitorStatsDataStream
            // 1. 添加水印和时间戳
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((vs, ts) -> vs.getTs())
            )
            // 2. 选取4个维度做为key:版本, 渠道, 地区, 新老用户标识
            .keyBy(vs -> vs.getVc() + "_" + vs.getCh() + "_" + vs.getAr() + "_" + vs.getIs_new())
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            // 3. 聚合
            .reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats stats1,
                                               VisitorStats stats2) throws Exception {
                        stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                        stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                        stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                        stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                        stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                        return stats1;
                    }
                },
                new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                    
                    private SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    
                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<VisitorStats> elements,
                                        Collector<VisitorStats> out) throws Exception {
                        final VisitorStats stats = elements.iterator().next();
                        // 补充统计时间
                        stats.setStt(f.format(new Date(context.window().getStart())));
                        stats.setEdt(f.format(new Date(context.window().getEnd())));
                        out.collect(stats);
                    }
                });
        return result;
    }
    
    /**
     * 解析出来3个流, 并把它们的类型map成一致的VisitorStats
     *
     * @param sourceStreams
     * @return
     */
    private DataStream<VisitorStats> parseStreamsAndUnionOne(Map<String, DataStreamSource<String>> sourceStreams) {
        // 1. 获取到3个流
        final DataStreamSource<String> uvStream = sourceStreams.get("dwm_uv");
        final DataStreamSource<String> pageLogStream = sourceStreams.get("dwd_page_log");
        final DataStreamSource<String> userJumpStream = sourceStreams.get("dwm_user_jump_detail");
        
        // 2. 若要把3个流合并为一个流需使用union, 所以需要把它们都转成同一类型, 然后才能union
        final SingleOutputStreamOperator<VisitorStats> uvStatsStream = uvStream
            .map(json -> {
                JSONObject jsonObj = JSON.parseObject(json);
                final JSONObject commonObj = jsonObj.getJSONObject("common");
                return new VisitorStats("", "",
                                        commonObj.getString("vc"),
                                        commonObj.getString("ch"),
                                        commonObj.getString("ar"),
                                        commonObj.getString("is_new"),
                                        1L, 0L, 0L, 0L, jsonObj.getJSONObject("page").getLong("during_time"),
                                        jsonObj.getLong("ts"));
            });
        
        final SingleOutputStreamOperator<VisitorStats> pageStatsStream = pageLogStream
            .map(json -> {
                final JSONObject jsonObj = JSON.parseObject(json, JSONObject.class);
                final JSONObject commonObj = jsonObj.getJSONObject("common");
                return new VisitorStats("", "",
                                        commonObj.getString("vc"),
                                        commonObj.getString("ch"),
                                        commonObj.getString("ar"),
                                        commonObj.getString("is_new"),
                                        0L, 1L, 0L, 0L, 0L,
                                        jsonObj.getLong("ts"));
            });
        
        final SingleOutputStreamOperator<VisitorStats> svPageStream = pageLogStream
            .flatMap(new FlatMapFunction<String, VisitorStats>() {
                @Override
                public void flatMap(String json, Collector<VisitorStats> out) throws Exception {
                    JSONObject jsonObj = JSON.parseObject(json);
                    final JSONObject commonObj = jsonObj.getJSONObject("common");
                    final String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        
                        final VisitorStats visitorStats = new VisitorStats("", "",
                                                                           commonObj.getString("vc"),
                                                                           commonObj.getString("ch"),
                                                                           commonObj.getString("ar"),
                                                                           commonObj.getString("is_new"),
                                                                           0L, 0L, 1L, 0L, 0L,
                                                                           jsonObj.getLong("ts"));
                        out.collect(visitorStats);
                    }
                }
            });
        
        final SingleOutputStreamOperator<VisitorStats> userJumpStatsStream = userJumpStream.map(json -> {
            final JSONObject jsonObj = JSON.parseObject(json, JSONObject.class);
            final JSONObject commonObj = jsonObj.getJSONObject("common");
            return new VisitorStats("", "",
                                    commonObj.getString("vc"),
                                    commonObj.getString("ch"),
                                    commonObj.getString("ar"),
                                    commonObj.getString("is_new"),
                                    0L, 0L, 0L, 1L, 0L,
                                    jsonObj.getLong("ts"));
        });
        
        return uvStatsStream.union(pageStatsStream, svPageStream, userJumpStatsStream);
    }
}
